package com.fwmagic.flink.projectcase.globalremoveduplication;

import com.fwmagic.flink.projectcase.globalremoveduplication.bean.Activity;
import com.fwmagic.flink.projectcase.utils.FlinkUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * uid全局去重之BloomFilter和Long实现
 * keyedState:
 *      ValueStateDescriptor<BloomFilter>
 *      ValueStateDescriptor<Long>
 * 1:曝光 2:点击 3:参与
 * 计算活动的点击人数、次数
 * A1，点击的次数：3 人数：2
 *
 * u001,A1,2019-09-02 10:10:11,1,北京市
 * u002,A1,2019-09-02 10:11:11,1,辽宁省
 * u001,A1,2019-09-02 10:11:11,2,北京市
 * u001,A1,2019-09-02 10:11:30,3,北京市
 * u002,A1,2019-09-02 10:12:11,2,辽宁省
 * u003,A2,2019-09-02 10:13:11,1,山东省
 * u003,A2,2019-09-02 10:13:20,2,山东省
 * u001,A1,2019-09-02 11:11:11,2,北京市
 */
public class ActivityCountV2Application {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = FlinkUtils.getEnv();

        env.setParallelism(4);

        ParameterTool parameters = ParameterTool.fromPropertiesFile(args[0]);

        DataStream<String> kafkaSource = FlinkUtils.createKafkaSource(parameters, SimpleStringSchema.class);


        //手动控制异常，验证出现异常，程序重启后，会不会重复消费数据
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);
        lines.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                if(value.startsWith("error")){
                    int i =1/0;
                }
                return value;
            }
        }).print();

        SingleOutputStreamOperator<Activity> maped = kafkaSource.map(new MapFunction<String, Activity>() {
            @Override
            public Activity map(String line) throws Exception {
                String[] fields = line.split(",");
                String uid = fields[0];
                String aid = fields[1];
                String time = fields[2].split(" ")[0];
                Integer type = Integer.parseInt(fields[3]);
                String province = fields[4];
                return Activity.of(uid, aid, time, type, province, 1);
            }
        });

        //按照指定条件分组，统计每个用户参与各类活动的次数
        //SingleOutputStreamOperator<Activity> result = maped.keyBy("aid",  "type").sum("count");
        //result.print();

        //统计不同活动[1:曝光 2:点击 3:参与]的次数
        KeyedStream<Activity, Tuple> keyed = maped.keyBy("aid", "type");

        keyed.print();

        keyed.map(new RichMapFunction<Activity, Tuple3<String, Integer, Long>>() {

            private transient ValueState<BloomFilter> uidsState;

            private transient ValueState<Long> countState;


            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<BloomFilter> valueStateDescriptor = new ValueStateDescriptor<BloomFilter>("uid-count-bloom-State", BloomFilter.class);
                ValueStateDescriptor<Long> countStateDescriptor = new ValueStateDescriptor<Long>("uid-count-long-State", Long.class);
                //初始化状态 OR 获取状态，用于存储去重的用户，以便统计用户的人数
                //用于去重
                uidsState = getRuntimeContext().getState(valueStateDescriptor);
                //用于计数(统计)
                countState = getRuntimeContext().getState(countStateDescriptor);
            }

            @Override
            public Tuple3<String, Integer, Long> map(Activity bean) throws Exception {
                String uid = bean.uid;
                BloomFilter bloomFilter = uidsState.value();
                if (bloomFilter == null) {
                    bloomFilter = BloomFilter.create(Funnels.unencodedCharsFunnel(), 1000000);
                    countState.update(0L);
                }
                Long counts = countState.value();
                if(!bloomFilter.mightContain(uid)){
                    //将当前用户加入到bloomFilter中
                    bloomFilter.put(uid);
                    //更新计数状态
                    countState.update(counts+=1);
                }

                //更新去重状态
                uidsState.update(bloomFilter);
                return Tuple3.of(bean.aid, bean.type, counts);
            }
        }).print();

        env.execute("ActivityCountV1Application");
    }
}
